package rabbit.flt.test.webflux.reactor;

import junit.framework.TestCase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rabbit.flt.common.context.TraceContext;
import rabbit.flt.common.trace.TraceData;
import rabbit.flt.plugins.common.ControllerPathHelper;
import rabbit.flt.test.common.spi.TestTraceHandler;
import rabbit.flt.test.webflux.BootController;
import rabbit.flt.test.webflux.ChildController;
import reactor.netty.http.client.HttpClient;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;

public class ReactorFrameworkCase {

    private Logger logger = LoggerFactory.getLogger(getClass());

    public void pathResolveCase() {
        Map<Class, List<String>> cache = ControllerPathHelper.getClassPathCache();
        TestCase.assertTrue(cache.containsKey(ChildController.class));
        TestCase.assertEquals(8, cache.get(ChildController.class).size());
        cache.get(ChildController.class).forEach(path -> TestCase.assertTrue(path.startsWith("/child")));

        TestCase.assertTrue(cache.containsKey(BootController.class));
        TestCase.assertEquals(6, cache.get(BootController.class).size());
        cache.get(BootController.class).forEach(path -> TestCase.assertTrue(path.startsWith("/mvc")));
    }

    public void reactorCallCase() throws InterruptedException {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logger.info("reactorCallCase: {}#{}, cost: {}", d.getNodeName(), d.getSpanId(), d.getCost());
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        intercept();
        semaphore.acquire(1);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/child/intercept"));
        TestCase.assertTrue(map.get("0").isHasMappedController());
        TestCase.assertEquals("unauthorized", map.get("0").getHttpResponse().getBody());
        TestCase.assertEquals(401, map.get("0").getHttpResponse().getStatusCode());
        logger.info("用例【正常捕获未进入controller的请求】验证成功");

        map.clear();
        call("/reactorSleep");
        semaphore.acquire(3);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/reactorSleep"));
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".reactorSleep"));
        TestCase.assertTrue(map.get("0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".reactorSleep"));
        TestCase.assertTrue(map.get("0-0-0").getCost() >= 10);
        TestCase.assertTrue(map.get("0-0").getCost() >= 10 && map.get("0-0").getCost() >= map.get("0-0-0").getCost());
        TestCase.assertTrue(map.get("0").getCost() >= 10 && map.get("0").getCost() >= map.get("0-0").getCost());
        logger.info("用例【controller和service都是reactor方法】验证成功");

        map.clear();
        call("/normalSleep");
        semaphore.acquire(3);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/normalSleep"));
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".normalSleep"));
        TestCase.assertTrue(map.get("0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".normalSleep"));
        TestCase.assertTrue(map.get("0-0-0").getCost() >= 10);
        TestCase.assertTrue(map.get("0-0").getCost() >= 10 && map.get("0-0").getCost() >= map.get("0-0-0").getCost());
        TestCase.assertTrue(map.get("0").getCost() >= 10 && map.get("0").getCost() >= map.get("0-0").getCost());
        logger.info("用例【controller是reactor方法，service是非reactor方法】验证成功");

        map.clear();
        call("/normal");
        semaphore.acquire(3);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/normal"));
        TestCase.assertNotNull(map.get("0").getHttpResponse());
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".normal"));
        TestCase.assertTrue(map.get("0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".normalSleep"));
        TestCase.assertTrue(map.get("0-0-0").getCost() >= 10);
        TestCase.assertTrue(map.get("0-0").getCost() >= 10 && map.get("0-0").getCost() >= map.get("0-0-0").getCost());
        TestCase.assertTrue(map.get("0").getCost() >= 10 && map.get("0").getCost() >= map.get("0-0").getCost());
        logger.info("用例【controller和service都是非reactor方法】验证成功");

        map.clear();
        call("/multiCall");
        semaphore.acquire(5);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/multiCall"));
        TestCase.assertNotNull(map.get("0").getHttpResponse());
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".multiCall"));
        TestCase.assertTrue(map.get("0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".multiCall"));
        TestCase.assertTrue(map.get("0-0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".reactorSleep"));
        TestCase.assertTrue(map.get("0-0-0-1").getNodeDesc().equals(ReactorService.class.getName() + ".reactorSleep"));
        TestCase.assertTrue(map.get("0-0-0").getCost() >= 20);
        TestCase.assertTrue(map.get("0-0").getCost() >= 20 && map.get("0-0").getCost() >= map.get("0-0-0").getCost());
        TestCase.assertTrue(map.get("0").getCost() >= 20 && map.get("0").getCost() >= map.get("0-0").getCost());
        logger.info("用例【多reactor方法flatMap调用】验证成功");

        map.clear();
        call("/multiFluxCall");
        semaphore.acquire(5);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/multiFluxCall"));
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".multiFluxCall"));
        TestCase.assertTrue(map.get("0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".multiFluxCall"));
        TestCase.assertTrue(map.get("0-0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".reactorSleep"));
        TestCase.assertTrue(map.get("0-0-0-1").getNodeDesc().equals(ReactorService.class.getName() + ".reactorSleep"));
        TestCase.assertTrue(map.get("0-0-0-0").getCost() >= 10);
        TestCase.assertTrue(map.get("0-0-0-1").getCost() >= 10);
        TestCase.assertTrue(map.get("0-0-0").getCost() >= 20);
        TestCase.assertTrue(map.get("0-0").getCost() >= 20 && map.get("0-0").getCost() >= map.get("0-0-0").getCost());
        TestCase.assertTrue(map.get("0").getCost() >= 20 && map.get("0").getCost() >= map.get("0-0").getCost());
        logger.info("用例【多reactor方法级联调用】验证成功");

        map.clear();
        call("/monoError");
        semaphore.acquire(3);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/monoError"));
        TestCase.assertEquals("reactor error", map.get("0").getHttpResponse().getBody());
        TestCase.assertEquals(500, map.get("0").getHttpResponse().getStatusCode());
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".monoError"));
        TestCase.assertTrue(map.get("0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".monoError"));
        TestCase.assertTrue(map.get("0-0-0").getData().contains("rabbit.flt.common.exception.FltException: reactor error"));
        logger.info("用例【mono异常调用】验证成功");


        map.clear();
        call("/fluxError");
        semaphore.acquire(3);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/fluxError"));
        TestCase.assertEquals("reactor error", map.get("0").getHttpResponse().getBody());
        TestCase.assertEquals(500, map.get("0").getHttpResponse().getStatusCode());
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".fluxError"));
        TestCase.assertTrue(map.get("0-0-0").getNodeDesc().equals(ReactorService.class.getName() + ".fluxError"));
        TestCase.assertTrue(map.get("0-0-0").getData().contains("rabbit.flt.common.exception.FltException: reactor error"));
        logger.info("用例【flux异常调用】验证成功");


        map.clear();
        call("/fluxData");
        semaphore.acquire(2);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/fluxData"));
        TestCase.assertEquals(200, map.get("0").getHttpResponse().getStatusCode());
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".fluxData"));
        logger.info("用例【flux调用】验证成功");


        map.clear();
        call("/fluxData2");
        semaphore.acquire(2);
        TestCase.assertTrue(map.get("0").getNodeName().equals("/reactor/fluxData2"));
        TestCase.assertEquals(200, map.get("0").getHttpResponse().getStatusCode());
        TestCase.assertTrue(map.get("0-0").getNodeDesc().equals(ReactorController.class.getName() + ".fluxData2"));
        logger.info("用例【flux调用】验证成功");


        TestTraceHandler.setDiscardDataHandler(null);
    }

    private void call(String path) {
        String uri = "http://localhost:8888/reactor" + path;
        HttpClient.create().get().uri(uri)
                .responseSingle((httpClientResponse, byteBufMono) -> byteBufMono.asByteArray().map(String::new))
                .block();
    }


    private void intercept() {
        String uri = "http://localhost:8888/child/intercept";
        HttpClient.create().get().uri(uri)
                .responseSingle((httpClientResponse, byteBufMono) -> byteBufMono.asByteArray().map(String::new))
                .block();
    }
}
